package xyj.sparrow.client;

import io.netty.channel.*;
import lombok.extern.slf4j.Slf4j;
import xyj.sparrow.client.handler.IdleClientHandler;
import xyj.sparrow.client.handler.NettyClientHandler;
import xyj.sparrow.client.listener.ConnectListener;
import xyj.sparrow.common.netty.decoder.ProtostuffDecoder;
import xyj.sparrow.common.discovery.SparrowServiceInstance;
import xyj.sparrow.common.netty.encoder.ProtostuffEncoder;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import lombok.Data;
import xyj.sparrow.discovery.api.SparrowDiscovery;


import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.ReentrantLock;


/**
 * 客户端
 *
 * @author Jim
 * @since 2021/1/12 14:33
 **/
@Data
@Slf4j
public class NettyClient implements Runnable {
    private Channel channel;
    private NettyClientHandler nettyClientHandler;
    private static final ExecutorService runThread = Executors.newFixedThreadPool(1);
    private ReentrantLock reentrantLock = new ReentrantLock();
    private EventLoopGroup group;
    private AtomicBoolean isActive = new AtomicBoolean(false);
    private SparrowServiceInstance sparrowServiceInstance;
    @Resource
    private SparrowDiscovery sparrowDiscovery;
    private Bootstrap bootstrap;

    @PostConstruct
    public void init() throws Exception {
        log.info("NettyClient init...");
        sparrowDiscovery.addListener("services");
        NettyClient that = this;
        group = new NioEventLoopGroup();
        bootstrap = new Bootstrap();
        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) {
                        ch.pipeline().addLast("encoder", new ProtostuffEncoder());
                        ch.pipeline().addLast("decoder", new ProtostuffDecoder(1024 * 1024, 0, 4, 0, 4));
                        ch.pipeline().addLast(new IdleStateHandler(60, 0, 0));
                        ch.pipeline().addLast(new IdleClientHandler(that));
                        NettyClientHandler nettyClientHandler = new NettyClientHandler();
                        ch.pipeline().addLast(nettyClientHandler);
                        that.nettyClientHandler = nettyClientHandler;
                    }
                });
        startWithThread();
    }

    /**
     * 使用配置构造nettyClient对象
     */
    public NettyClient() {

    }

    /**
     * TODO
     *
     * @author 江南小俊
     * @since 2021/6/15 下午10:49
     **/
    @Override
    public void run() {

        try {
            //获取服务实例
            reacquireServiceInstance();
            ChannelFuture future = bootstrap.connect(sparrowServiceInstance.getAddress(),
                    sparrowServiceInstance.getPort());
            future.addListener(new ConnectListener(this));
            future.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            log.error("", e);
        }
    }

    /**
     * 重新获取服务实例
     *
     * @author 江南小俊
     * @since 2021/6/20 上午10:10
     **/


    private void reacquireServiceInstance() {
        try {
            sparrowServiceInstance = sparrowDiscovery.getInstance("services");
            while (sparrowServiceInstance == null) {
                sparrowServiceInstance = sparrowDiscovery.getInstance("services");
                Thread.sleep(1000);
            }
        } catch (Exception e) {
            log.error("", e);
        }
    }

    @PreDestroy
    public void destroy() {
        group.shutdownGracefully();
    }

    public void startWithThread() {
        log.info("NettyClientThread start ....");
        runThread.execute(this);
    }

    public void connect() {
        reentrantLock.lock();
        if (!isActive.get()) {
            run();
        }
        reentrantLock.unlock();
    }


}
